package util;

import data.DataPoint;
import org.apache.beam.sdk.transforms.DoFn;
import org.apache.hadoop.hbase.client.Mutation;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.util.Bytes;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.nio.charset.Charset;

public class HbaseMap extends DoFn<DataPoint, Mutation> {

    private static final Logger LOG = LoggerFactory.getLogger(HbaseMap.class);

    @ProcessElement
    public void processElement(ProcessContext c) {

        byte[] qual = Bytes.toBytes(c.element().sensorId);
        byte[] cf = Bytes.toBytes("v");
        byte[] row = Bytes.toBytes("0_" + c.element().timestamp);
        byte[] val = Bytes.toBytes(c.element().timestamp);
        final Charset UTF_8 = Charset.forName("UTF-8");
        Mutation mutation = new Put(row).addColumn(cf, qual, val);
        c.output(mutation);





    }
}
